package com.atguigu.sparkcore.rdd.file.bbase

import com.atguigu.sparkcore.util.MySparkContextUtil
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{Cell, CellUtil}
import org.apache.spark.rdd.RDD

/**
 * description ：读取hbase中的数据
 * author ：剧情再美终是戏
 * mail : 13286520398@163.com
 * date ：Created in 2020/1/9 10:48
 * modified By ：
 * version: : 1.0
 */
object RHbase {

  def main(args: Array[String]): Unit = {

    // 创建sparkContext
    val sc = MySparkContextUtil.get(args)

    // 创建hbase相关配置
    val conf = new Configuration()
    conf.set("hbase.zookeeper.quorum", "hadoop101,hadoop102,hadoop103")
    conf.set("hbase.mapreduce.inputtable", "user")

    // 读取hbase表中的数据
    val habseRdd: RDD[(ImmutableBytesWritable, Result)] = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])

    // 输出信息
    habseRdd.foreach { case (keys, result) => val key = Bytes.toString(keys.get())
      // TODO 1、使用scala中的方法（rawCells），并也使用scala方法遍历
      //val cells: Array[Cell] = result.rawCells()

      // TODO 2、使用java中的方法(listCells),使用scala方法遍历，需要隐式转换
      import scala.collection.JavaConversions._
      val cells = result.listCells()
      for (cell <- cells) {
        val family = Bytes.toString(CellUtil.cloneFamily(cell))
        val cloumn = Bytes.toString(CellUtil.cloneQualifier(cell))
        val value = Bytes.toString(CellUtil.cloneValue(cell))
        printf("key=%s, family=%s, cloumn=%s, value=%s\n", key, family, cloumn, value)
      }
    }

    // 关闭sparkContext资源
    MySparkContextUtil.close(sc)
  }


}
